#include "sqlfs.h"
#include "log.h"
#include "config.h"
#include <iostream>
#include <fuse/fuse.h>
#include <mysql++/mysql++.h>
#include <string>
#include <libgen.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include "pthread.h"
#include <signal.h>
#include "serializer.h"
#include "offset_table.h"
#include "logparser.h"
#include "replication_ignore.h"
#include "monitor.h"
using namespace std;
using namespace mysqlpp;
#ifdef DEBUG
#define catchex(x) {\
	/*LOG << ENDL;*/\
	int res=-1;\
	try { res=(x); }\
	catch(Exception e) { LOG_ERR << e.what() << ENDL; }\
	catch(...){ LOG_ERR << "something went terribly wrong" << ENDL;}\
	return res;\
}
#else
#define catchex(x){return (x);}
#endif

char mount_dir[500];
enum Log_event_type
{
  /*
    Every time you update this enum (when you add a type), you have to
    fix Format_description_log_event::Format_description_log_event().
  */
  UNKNOWN_EVENT= 0,
  START_EVENT_V3= 1,
  QUERY_EVENT= 2,
  STOP_EVENT= 3,
  ROTATE_EVENT= 4,
  INTVAR_EVENT= 5,
  LOAD_EVENT= 6,
  SLAVE_EVENT= 7,
  CREATE_FILE_EVENT= 8,
  APPEND_BLOCK_EVENT= 9,
  EXEC_LOAD_EVENT= 10,
  DELETE_FILE_EVENT= 11,
  /*
    NEW_LOAD_EVENT is like LOAD_EVENT except that it has a longer
    sql_ex, allowing multibyte TERMINATED BY etc; both types share the
    same class (Load_log_event)
  */
  NEW_LOAD_EVENT= 12,
  RAND_EVENT= 13,
  USER_VAR_EVENT= 14,
  FORMAT_DESCRIPTION_EVENT= 15,
  XID_EVENT= 16,
  BEGIN_LOAD_QUERY_EVENT= 17,
  EXECUTE_LOAD_QUERY_EVENT= 18,

  TABLE_MAP_EVENT = 19,

  /*
    These event numbers were used for 5.1.0 to 5.1.15 and are
    therefore obsolete.
   */
  PRE_GA_WRITE_ROWS_EVENT = 20,
  PRE_GA_UPDATE_ROWS_EVENT = 21,
  PRE_GA_DELETE_ROWS_EVENT = 22,

  /*
    These event numbers are used from 5.1.16 and forward
   */
  WRITE_ROWS_EVENT = 23,
  UPDATE_ROWS_EVENT = 24,
  DELETE_ROWS_EVENT = 25,

  /*
    Something out of the ordinary happened on the master
   */
  INCIDENT_EVENT= 26,

  /*
    Add new events here - right above this comment!
    Existing events (except ENUM_END_EVENT) should never change their numbers
  */

  ENUM_END_EVENT /* end marker */
};



pthread_t thread_fuse, thread_binlog;
static SqlFs *fs;
//First the stuff that gets used:
static int my_getattr (const char *path, struct stat *stbuf) {
    catchex(fs->getattr (path, stbuf));
}
static int my_readdir (const char *path, void *buf, fuse_fill_dir_t filler,
	      off_t offset, struct fuse_file_info *fi) {
    catchex(fs->readdir (path, buf, filler, offset, fi));
}
static int my_open (const char *path, struct fuse_file_info *fi) {
    catchex(fs->open (path, fi));
}

static int my_read (const char *path, char *buf, size_t size, off_t offset,
	   struct fuse_file_info *fi) {
    catchex(fs->read (path, buf, size, offset, fi));
}

static int my_write (const char *path, const char *buf, size_t size,
	    off_t offset, struct fuse_file_info *fi) {
    catchex(fs->write (path, buf, size, offset, fi));
}
static int my_release (const char *path, struct fuse_file_info *fi) {
    catchex(fs->release (path, fi));
}
static int my_flush(const char* path, struct fuse_file_info* fi) {
	catchex(fs->flush(path, fi));
}

static void* my_init(struct fuse_conn_info* conn) {
	return fs->init(conn);
}

static void my_destroy(void* conn) {
	fs->destroy(conn);
}
static int my_opendir(const char* path, struct fuse_file_info* fi) {
	catchex(fs->opendir(path, fi));
}
static int my_releasedir(const char* path, struct fuse_file_info* fi) {
	catchex(fs->releasedir(path, fi));
}
//the stuff that's currently not implemented, or I could not find a meaningful way to implement:
static int my_access (const char *path, int mask) {
    catchex(fs->access (path, mask));
}
static int my_mknod (const char *path, mode_t mode, dev_t rdev) {
    catchex(fs->mknod (path, mode, rdev));
}
static int my_mkdir (const char *path, mode_t mode) {
   catchex(fs->mkdir (path, mode));
}
static int my_unlink (const char *path) {
    catchex(fs->unlink (path));
}
static int my_rmdir (const char *path) {
    catchex(fs->rmdir (path));
}
static int my_symlink (const char *from, const char *to) {
    catchex(fs->symlink (from, to));
}
static int my_rename (const char *from, const char *to) {
    catchex(fs->rename (from, to));
}
static int my_link (const char *from, const char *to) {
    catchex(fs->link (from, to));
}
static int my_chmod (const char *path, mode_t mode) {
    catchex(fs->chmod (path, mode));
}
static int my_chown (const char *path, uid_t uid, gid_t gid) {
    catchex(fs->chown (path, uid, gid));
}
static int my_truncate (const char *path, off_t size) {
    catchex(fs->truncate (path, size));
}
static int my_utimens (const char *path, const timespec ts[]) {
    catchex(fs->utimens (path, ts));
}
static int my_statfs (const char *path, struct statvfs *stbuf) {
    catchex(fs->statfs (path, stbuf));
}
static int my_fsync (const char *path, int isdatasync, struct fuse_file_info *fi) {
    catchex(fs->fsync (path, isdatasync, fi));
}

static int my_fsyncdir(const char* path, int datasync, struct fuse_file_info* fi) {
	catchex(fs->fsyncdir(path, datasync, fi));
}
static int my_readlink(const char* path, char* target, size_t size) {
	catchex(fs->readlink(path, target, size));
}
#ifdef HAVE_SETXATTR
static int my_setxattr (const char *path, const char *name, const char *value,
	       size_t size, int flags) {
    catchex(fs->setxattr (path, name, value, size, flags));
}

static int my_getxattr (const char *path, const char *name, char *value, size_t size) {
    catchex(fs->getxattr (path, name, value, size));
}

static int my_listxattr (const char *path, char *list, size_t size) {
    catchex(fs->listxattr (path, list, size));
}

static int my_removexattr (const char *path, const char *name) {
    catchex(fs->removexattr (path, name));
}
#endif				/* HAVE_SETXATTR */
struct my_fuse_operations: fuse_operations {
    my_fuse_operations() {
      getattr = my_getattr;
      access = my_access;
      readlink = my_readlink;
      readdir = my_readdir;
      mknod = my_mknod;
      mkdir = my_mkdir;
      unlink = my_unlink;
      rmdir = my_rmdir;
      symlink = my_symlink;
      rename = my_rename;
      link = my_link;
      chmod = my_chmod;
      chown = my_chown;
      truncate = my_truncate;
      utimens = my_utimens;
      open = my_open;
      read = my_read;
      write = my_write;
      statfs = my_statfs;
      release = my_release;
      fsync = my_fsync;
	  flush = my_flush;
	  opendir = my_opendir;
	  releasedir = my_releasedir;
	  fsyncdir = my_fsyncdir;
	  init = my_init;
	  destroy = my_destroy;
#ifdef HAVE_SETXATTR
      setxattr = my_setxattr;
      getxattr = my_getxattr;
      listxattr = my_listxattr;
      removexattr = my_removexattr;
#endif
    }
} oper;

void find_binlog() {
	Connection* con = Pool::Get()->grab();
	DB_INIT;
	DB_START;
	Query q = con->query();
	q << "SHOW MASTER STATUS";
	UseQueryResult res = q.use();
	Row row = res.fetch_row();
	DB_STOP;
	binlog_file = strdup(row[0].c_str());
	binlog_offset = (unsigned long long)row[1];
	//printf("binlog_file=%s, binlog_offset=%llu\n",binlog_file, binlog_offset);
	Pool::Get()->release(con);
}

int update_db_index(string table, int ev_type, size_t rowid, vector<string>& vals1, vector<string>& vals2) {
	OffsetTable *t;
	TREE_INIT;
	switch(ev_type) {
	case CREATE_TABLE_EVENT:
		OffsetTable::Get(table);
		return 0;
	case DROP_TABLE_EVENT:
		OffsetTable::drop(table);
		return 0;
	case TRUNCATE_TABLE_EVENT:
		if(!repl_ignore::shall_skip_truncate(table)) {
			TREE_START;
			OffsetTable::Get(table)->clear();
			TREE_STOP;
		}
		return 0;
	}
	t = OffsetTable::Get(table);
	if(t == NULL) {
		error("table %s not found", table.c_str());
		exit(-1);
	}
	size_t rowSize = Serializer::RowSize(vals1, 1);
	switch(ev_type) {
	case UPDATE_ROWS_EVENT:
		//printf("update rows event");
//		printf("updating row %d size diff = %d\n",
//				rowid, (long long int)Serializer::RowSize(vals2, 1)
//						- (long long int)Serializer::RowSize(vals1, 1));
		TREE_START;
		t->updateRowSize(rowid,
							Serializer::RowSize(vals1, 1),
							Serializer::RowSize(vals2, 1));
		TREE_STOP;
		break;
	case DELETE_ROWS_EVENT:
		//printf("delete rows event\n");
//		printf("removing row %d of size %d\n", rowid, Serializer::RowSize(vals1, 1));
		if(!repl_ignore::shall_skip_delete(table, rowSize)) {
			TREE_START;
			t->updateRowSize(rowid, Serializer::RowSize(vals1,1), 0);
			TREE_STOP;
		}
		break;
	case WRITE_ROWS_EVENT:
		if(!repl_ignore::shall_skip_insert(table, rowSize)) {
			TREE_START;
			t->updateRowSize(rowid, 0, Serializer::RowSize(vals1, 1));
			TREE_STOP;
		}
		//printf("write rows event\n");
//		printf("inserting row %d of size %d\n", rowid, Serializer::RowSize(vals1, 1));
		break;
	default:
		error("unknown row event type %d", ev_type);
		return 1;
	}
//	t->print();
	return 0;
}

void* check_remote_log(void *arg) {
	Connection* con = Pool::Get()->grab();
	DB_INIT;
	Query q = con->query();
	(void)arg;
	find_binlog();
	unsigned long long new_offset = binlog_offset;
	char new_binlog_file[256];
	strcpy(new_binlog_file, binlog_file);
	for(;;) {
		if(binlog_offset >= new_offset && !strcmp(new_binlog_file, binlog_file)) {
			DB_START;
			q << "SHOW MASTER STATUS";
			UseQueryResult res = q.use();
			Row row = res.fetch_row();
			DB_STOP;
			strcpy(new_binlog_file, row[0].c_str());
			new_offset = (unsigned long long)row[1];
			q.reset();
			sleep(1);
		} else {
//			if(strcmp(binlog_file, new_binlog_file)) {
//
//			} else {
//				if(binlog_offset > new_offset) {
//					//fprintf(stderr, "binlog_offset=%llu, new_offset=%llu\n",binlog_offset, new_offset);
//				}
//			}
			get_remote_events(update_db_index);
		}
	}
	Pool::Get()->release(con);
	return NULL;
}

void* check_local_log(void *arg) {
	(void)arg;
	find_binlog();
	int fd = open(binlog_file, O_RDONLY);
	struct stat stbuf;
	fstat(fd, &stbuf);
	while(true) {
		fstat(fd, &stbuf);
		if(binlog_offset < (unsigned long long)stbuf.st_size) {
//			printf("log file has grown\n");
			parse_log(update_db_index);
		}
		sleep(1);
	}
	return NULL;
}
struct argst {
	int argc;
	char **argv;
};

void* start_fuse(void *arg) {
	struct argst *args = (struct argst*)arg;
	char **argv = args->argv;
	int argc = args->argc;
	string path = argv[argc-1];
	//LOG << "path = " << path << ENDL;
	string cmd = "mkdir -p ";
	cmd += path;
	int ret = system(cmd.c_str());
	(void)ret;
	cmd = "fusermount -u ";
	cmd += path;
	ret = system(cmd.c_str());
	(void)ret;
	size_t found = path.rfind('/');
	//LOG << "found = " << found << ENDL;
	string basedir = path.substr(0, found+1);
	basedir += ".";
	basedir += basename(argv[argc-1]);
	//LOG << "basedir = " << basedir << ENDL;
	cmd = "mkdir -p ";
	cmd += basedir;
	ret = system(cmd.c_str());
	(void)ret;
	ret = system("mkdir -p /tmp/sqlfslog");
	(void)ret;
	fs = SqlFs::Get();
	fs->setRootDir(basedir.c_str());
	char** fuse_cmd_opts = new char*[10];
	for(unsigned int i = 0; i < 10; ++i)
		fuse_cmd_opts[i] = new char[100];
	unsigned int fuse_argc = 0;
	strcpy(fuse_cmd_opts[fuse_argc++], argv[0]);
#ifndef MULTITHREADED_FUSE
	strcpy(fuse_cmd_opts[fuse_argc++], "-s");
#endif
	strcpy(fuse_cmd_opts[fuse_argc++], "-f");
	strcpy(fuse_cmd_opts[fuse_argc++], "-oallow_other");
	strcpy(fuse_cmd_opts[fuse_argc++], argv[argc-1]);

	fuse_main(fuse_argc, (char**)fuse_cmd_opts, &oper, NULL);
	return NULL;
}
void kill_handler(int signal) {
	pthread_kill(thread_fuse, SIGTERM);
	pthread_kill(thread_binlog, SIGTERM);
	sleep(1);
	string cmd = "fusermount -u ";
	cmd += mount_dir;
	int ret = system(cmd.c_str());
	(void)ret;
	DB_REPORT;
	TREE_REPORT;
	exit(0);
}

int main(int argc, char *argv[]) {
	struct sigaction new_act, old_act;
	new_act.sa_handler = kill_handler;
	sigemptyset (&new_act.sa_mask);
	new_act.sa_flags = 0;
	sigaction(SIGINT, &new_act, &old_act);
    log_init(LOG_FILE);
    struct argst args;
    args.argc = argc;
    args.argv = argv;
    int ho_error;
    binlog_offset = 0;
    if ((ho_error=handle_parse_args(&argc, &argv)))
    	    exit(ho_error);
//    printf("binlog_offset=%llu\n", binlog_offset);
//    printf("binlog_file=%s\n", binlog_file);
    strcpy(mount_dir, argv[argc-1]);
    repl_ignore::init();
    int iret1 = pthread_create( &thread_fuse, NULL, start_fuse, &args);
    int iret2 = pthread_create(&thread_binlog, NULL, check_remote_log, NULL);
    pthread_join(thread_fuse, NULL);
    pthread_join(thread_binlog, NULL);
    printf("thread_fuse returns: %d\n",iret1);
    printf("thread_binlog returns: %d\n",iret2);
    //fuse_opt_free_args(&args);
	log_finish();
    return EXIT_SUCCESS;
}

